package git.soulbgm.websocket;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import git.soulbgm.common.enums.Status;
import git.soulbgm.common.pojo.Response;
import git.soulbgm.common.enums.CodeEnum;
import git.soulbgm.pojo.AlarmInfo;
import git.soulbgm.pojo.NodeStatus;
import git.soulbgm.pojo.ServiceStatus;
import git.soulbgm.pojo.SoftwareStatus;
import git.soulbgm.service.AlarmInfoService;
import git.soulbgm.service.NodeStatusService;
import git.soulbgm.service.ServiceStatusService;
import git.soulbgm.service.SoftwareStatusService;
import git.soulbgm.utils.JsonTool;
import git.soulbgm.utils.LogHelper;
import git.soulbgm.utils.SpringBeanUtil;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static git.soulbgm.common.MonitorConstants.*;

/**
 * 监控相关WebSocket
 *
 * @author SoulBGM
 * @date 2023-04-17
 */
@ServerEndpoint("/monitor/{username}")
@Component
public class MonitorWebSocket {

    private static final LogHelper LOG = new LogHelper(MonitorWebSocket.class);
    private static final AtomicInteger ONLINE_NUM = new AtomicInteger();
    private static final ConcurrentHashMap<String, Session> SESSION_POOLS = new ConcurrentHashMap<>();

    private static NodeStatusService nodeStatusService;
    private static ServiceStatusService serviceStatusService;
    private static SoftwareStatusService softwareStatusService;
    private static AlarmInfoService alarmInfoService;

    /**
     * 发送消息
     *
     * @param session 会话
     * @param message 消息
     * @throws IOException IO异常
     */
    public static synchronized void sendMessage(Session session, String message) throws IOException {
        if (session != null) {
            session.getBasicRemote().sendText(message);
        }
    }

    /**
     * 给指定用户发送信息
     *
     * @param username 用户名
     * @param message  消息
     */
    public static void sendInfo(String username, String message) {
        Session session = SESSION_POOLS.get(username);
        try {
            sendMessage(session, message);
        } catch (Exception e) {
            LOG.error("给 {} 用户发送数据出现异常", username, e);
        }
    }

    /**
     * 给指定用户发送信息
     *
     * @param username 用户名
     * @param obj      消息对象
     */
    public static void sendInfo(String username, Object obj) {
        sendInfo(username, JsonTool.toJson(obj));
    }

    /**
     * 群发消息
     *
     * @param message 消息
     */
    public static void broadcast(String message) {
        Collection<Session> sessions = SESSION_POOLS.values();
        for (Session session : sessions) {
            try {
                sendMessage(session, message);
            } catch (Exception e) {
                LOG.error("群发消息出现异常", e);
            }
        }
    }

    /**
     * 群发消息
     *
     * @param obj 消息对象
     */
    public static void broadcast(Object obj) {
        broadcast(JsonTool.toJson(obj));
    }

    @OnOpen
    public void onOpen(Session session, @PathParam("username") String username) {
        SESSION_POOLS.put(username, session);
        addOnlineCount();
        initService();
        sendMonitorData(username);
    }

    @OnClose
    public void onClose(@PathParam("username") String username) {
        SESSION_POOLS.remove(username);
        cutOnlineCount();
    }

    @OnMessage
    public void onMessage(@PathParam("username") String username, String message) {

    }

    @OnError
    public void onError(Session session, Throwable throwable) {
        LOG.error("会话IP {} ,出现异常", session.getRequestURI().getHost(), throwable);
    }

    /**
     * 初始化调用服务
     */
    private static void initService() {
        if (nodeStatusService == null) {
            nodeStatusService = SpringBeanUtil.getBean(NodeStatusService.class);
        }
        if (serviceStatusService == null) {
            serviceStatusService = SpringBeanUtil.getBean(ServiceStatusService.class);
        }
        if (softwareStatusService == null) {
            softwareStatusService = SpringBeanUtil.getBean(SoftwareStatusService.class);
        }
        if (alarmInfoService == null) {
            alarmInfoService = SpringBeanUtil.getBean(AlarmInfoService.class);
        }
    }

    /**
     * 发送监控数据
     *
     * @param username 用户名
     */
    private static void sendMonitorData(String username) {
        Response data = getMonitorData();
        sendInfo(username, data);
    }

    /**
     * 获得监控数据
     *
     * @return {@link Response}
     */
    public static Response getMonitorData() {
        Response data = Response.setOk(CodeEnum.SUCCESS);
        Map<String, Object> map = new HashMap<>(4);
        map.put(SHOW_NODE, getNodeMap());
        map.put(SHOW_SERVICE, getServiceMap());
        map.put(SHOW_SOFTWARE, getSoftwareMap());
        map.put(SHOW_ALARM, getAlarmInfoList());
        data.setData(map);
        return data;
    }

    /**
     * 获取节点状态Map
     *
     * @return {@link Map}<{@link String}, {@link Long}>
     */
    private static Map<String, Long> getNodeMap() {
        Map<String, Long> map = nodeStatusService.list().stream().collect(Collectors.groupingBy(MonitorWebSocket::getStatus, Collectors.counting()));
        map.putIfAbsent(NODE_STATUS_NORMAL, 0L);
        map.putIfAbsent(NODE_STATUS_ABNORMAL, 0L);
        return map;
    }

    /**
     * 获得状态
     *
     * @param s 节点状态
     * @return {@link String}
     */
    private static String getStatus(NodeStatus s) {
        Integer status = s.getStatus();
        if (status.equals(Status.NORMAL.code) || status.equals(Status.STOPPING.code)) {
            return NODE_STATUS_NORMAL;
        } else {
            return NODE_STATUS_ABNORMAL;
        }
    }

    /**
     * 获取服务状态Map
     *
     * @return {@link Map}<{@link String}, {@link List}<{@link ServiceStatus}>>
     */
    private static Map<String, List<ServiceStatus>> getServiceMap() {
        return serviceStatusService.list().stream().collect(Collectors.groupingBy(ServiceStatus::getServiceName));
    }

    /**
     * 获取软件状态Map
     *
     * @return {@link Map}<{@link String}, {@link List}<{@link SoftwareStatus}>>
     */
    private static Map<String, List<SoftwareStatus>> getSoftwareMap() {
        return softwareStatusService.list().stream().collect(Collectors.groupingBy(SoftwareStatus::getSoftwareName));
    }

    /**
     * 获得最近50条没有处理的告警信息列表
     *
     * @return {@link List}<{@link AlarmInfo}>
     */
    private static List<AlarmInfo> getAlarmInfoList() {
        Page<AlarmInfo> page = alarmInfoService.page(new Page<>(1, 50, false),
                new LambdaQueryWrapper<AlarmInfo>().eq(AlarmInfo::getArtificialHandle, HANDLE_NO).orderByDesc(AlarmInfo::getCreateTime));
        return page.getRecords();
    }

    /**
     * 添加在线计数
     */
    public static void addOnlineCount() {
        ONLINE_NUM.incrementAndGet();
    }

    /**
     * 减少在线计数
     */
    private static void cutOnlineCount() {
        ONLINE_NUM.decrementAndGet();
    }

    /**
     * 获得在线数量
     *
     * @return {@link AtomicInteger}
     */
    public static AtomicInteger getOnlineNum() {
        return ONLINE_NUM;
    }

    /**
     * 获得会话池
     *
     * @return {@link ConcurrentHashMap}<{@link String}, {@link Session}>
     */
    public static ConcurrentHashMap<String, Session> getSessionPools() {
        return SESSION_POOLS;
    }
}
